package generic.activemq;

import generic.Publisher;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.JMSException;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;
import javax.jms.Topic;

import org.apache.activemq.ActiveMQConnectionFactory;

public class ActiveMqPublisher implements Publisher {

	Session session;
	MessageProducer messageProducer;

	//private String url = "failover:(tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61616,tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61618)?randomize=false";
	private String url = "failover:(tcp://192.168.11.128:61616,tcp://192.168.11.129:61616)?randomize=false&backup=true&initialReconnectDelay=100&maxReconnectDelay=100&useExponentialBackOff=false&maxReconnectAttempts=1&timeout=100";
	//private String url = "failover:(tcp://192.168.11.128:61616,tcp://192.168.11.129:61616)?randomize=false";
	//private String url = "fanout:(static:(tcp://192.168.11.128:61616,tcp://192.168.11.129:61616))?minAckCount=1";
	//private String url = "fanout:(static:(tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61616,tcp://ec2-54-242-185-60.compute-1.amazonaws.com:61618))?minAckCount=1";
	// private String url = "tcp://localhost:61616";
	//private String url = "failover:(tcp://localhost:61616,tcp://localhost:61618)?randomize=false&backup=false&initialReconnectDelay=100&maxReconnectAttempts=1";
	// private String url =
	// "fanout:(static:(tcp://localhost:61616,tcp://localhost:61618))?minAckCount=1&initialReconnectDelay=1&maxReconnectDelay=1&maxReconnectAttempts=1";

	private StringBuilder builder;
	private boolean transacted = true;
	private boolean persistent = true;

	public ActiveMqPublisher(String url) {
		super();
		this.url = url;
		System.out.println("url:" + url);

		builder = new StringBuilder();

		initialize(url);
		
		System.out.println("ActiveMqPublisher ready to publish messages...");
	}
	
	private void initialize(String brokerUrl) {
		try {
			ActiveMQConnectionFactory factory = new ActiveMQConnectionFactory(
					brokerUrl);
			Connection connection = factory.createConnection();
			//connection.setClientID("publisher1");
			// ((ActiveMQConnection)connection).setUseAsyncSend(false);
			// ((ActiveMQConnection)connection).setAlwaysSessionAsync(false);
			// ((ActiveMQConnection)connection).setAlwaysSyncSend(true);
			session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
			Topic topic = session.createTopic("topictest.messages");

			messageProducer = session.createProducer(topic);
			if( persistent )
				messageProducer.setDeliveryMode(DeliveryMode.PERSISTENT);
			else
				messageProducer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);

			connection.start();
		} catch (JMSException e) {
			e.printStackTrace();
		}
	}

	@Override
	public boolean send(String message) {
		for( int retryCount = 0 ; (retryCount < 2) ; retryCount++ )  {
			try {
				TextMessage textMessage = session.createTextMessage(message);
				messageProducer.send(textMessage);
				if (transacted) {
					session.commit();
				}
				return true;
			} 
			catch (javax.jms.TransactionRolledBackException e) {
				System.out.println(e.toString());
				// We just try again !!
			}
			catch (JMSException e) {
				e.printStackTrace();
				return false;
			}
		}
		return false;
	}

}
